package com.ds.lens.flink.monitor.monitorjob.function;

import com.ds.lens.data.common.constant.Constants;
import com.ds.lens.flink.monitor.common.bo.MonitorDefinition;
import com.ds.lens.flink.monitor.common.MonitorDefinitionFactory;
import com.ds.lens.flink.monitor.common.LogUtils;
import com.ds.lens.flink.monitor.common.bo.MetricBundle;
import com.ds.lens.flink.monitor.common.bo.MonitorState;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class MetricRetriveFunction extends ProcessWindowFunction<MonitorState, MetricBundle, Long, TimeWindow> {

    @Override
    public void process(Long key, Context context, Iterable<MonitorState> elements, Collector<MetricBundle> out) throws Exception {
        try {
            Map<Long, MonitorState> map = new HashMap<>();
            for (MonitorState state : elements) {
                map.put(state.getKey().getMonitorId(), state);
            }
            long windowStartTime = context.window().getStart();
            long windowEndTime = context.window().getEnd();
            for (Long monitorId : map.keySet()) {
                MonitorState state = map.get(monitorId);
                MonitorDefinition definition = MonitorDefinitionFactory.get().getMonitorDefinitionByState(state);
                if (definition == null) {
                    continue;
                }
                //检查monitor是否已经被运维事件挂起
                if (state.getRunStatus() == Constants.MonitorRunStatusEnum.SUSPENDED.getCode()) {
                    continue;
                }
                long interval = definition.getIntervalTime().longValue();
                long endTime = windowStartTime - definition.getRetriveDelayTime() * 60000;
                if (endTime % (interval * 60000) != 0) {
                    continue;
                }
                long startTime = endTime - definition.getRetiveDuration() * 60000;
                MetricBundle bundle = MonitorDefinitionFactory.get().getMetricBundleByState(state, startTime, endTime, interval);
                bundle.setWindowStartTime(windowStartTime);
                bundle.setWindowEndTime(windowEndTime);
                LogUtils.debug("collect metricbundle={}", bundle);
                out.collect(bundle);
            }
        } catch (Exception e) {
            log.error("error occurred", e);
        }
    }

//    public static void main(String[] args) {
//        System.out.println(stringToMD5("123"));
//        System.out.println(stringToMD5("11111111111111111111"));
//        System.out.println(stringToMD5("4444444"));
//        System.out.println(stringToMD5("sadfasdfasdfasdfasdfasdfsadfsdfsadfasdfasdfasdfasdfasdfsadfsdfsadfasdfasdfasdfasdfasdfsadfsdfsadfasdfasdfasdfasdfasdfsadfsdf"));
//        System.out.println(stringToMD5("s"));
//        System.out.println(stringToMD5("biz=\"HLL/TC/INF/AIOPS\",cluster=\"apm-online\",consul_service=\"ump-business_target_node\",hostname=\"apm-es-client001\",idc=\"ali-bj\",instance=\"172.23.58.31:9200\",job=\"ump-business_target_node\",monitor_idc=\"sj02\",node=\"apm-es-client001\",nodeid=\"G5Jz-WmkSXefY_uDTNdEng\",svc=\"ump-business_target_node\",ump_biz=\"HLL/TC/INF/AIOPS\",ump_hostname=\"apm-es-client001\",ump_idc=\"ali-bj\",ump_metrics_path_al=\"_prometheus.metrics\",ump_monitor_idc=\"sj02\",ump_svc=\"apm\""));
//    }
//    public static String stringToMD5(String plainText) {
//        byte[] secretBytes = null;
//        try {
//            secretBytes = MessageDigest.getInstance("md5").digest(
//                    plainText.getBytes());
//        } catch (NoSuchAlgorithmException e) {
//            throw new RuntimeException("没有这个md5算法！");
//        }
//        String md5code = new BigInteger(1, secretBytes).toString(16);
//        for (int i = 0; i < 32 - md5code.length(); i++) {
//            md5code = "0" + md5code;
//        }
//        return md5code;
//    }

}
